package com.example.rxjavademo;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import com.example.rxjavademo.Net.RetrofitProxy;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BooleanSupplier;
import io.reactivex.functions.Cancellable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {

    private Observable<String> observable;
    private Observable<String> justObservable;
    private Observable<List<String>> justStrObservable;
    private Observable<String> iterableObervable;
    private Observable<String> deferObservable;
    private Observable<Long> intervalObservavble;
    private Observable<Integer> rangeObservable;
    private Observable<Long> timerObservable;
    private Observable<String> repeatObservable;

    private Observer<String> observer;
    private Consumer<String> consumer;
    private Consumer<Integer> consumer2;
    private Function function;

    private CompositeDisposable compositeDisposable;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        compositeDisposable = new CompositeDisposable();

        createObservable();
        createFlowable();
        createObserver();
        listActionType();
        listOperators();
        requireScheduler();


        requestData();
        applyTransformer();
        //subscribeEvent();
    }

    private void applyTransformer() {
        Disposable disposable = justObservable.compose(TransformerUtils.applyTransformer())
                .compose(TransformerUtils.<Integer>observableToMain())
                .subscribe(consumer2);
    }

    /**
     * create创建的方式：在ObservableOnSubscribe内部初始化数据后，通知Observer
     */
    private void createObservable() {
        observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) { //当被观察者被订阅，会触发该方法，以分发事件方式通知观察者
                //1.执行操作
                //2.触发回调，通知观察者
                emitter.onNext("create way");
            }
        });

        justObservable = Observable.just("just way", "just2");

        List<String> arrayList = new ArrayList<>();
        arrayList.add("just1");
        arrayList.add("just2");
        arrayList.add("just");
        justStrObservable = Observable.just(arrayList);

        ArrayList<String> list = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            list.add("iterable data_" + i);
        }
        iterableObervable = Observable.fromIterable(list); // call list.size() of onNext() method, each time emit an item

        //延迟创建被观察者
        deferObservable = Observable.defer(new Callable<ObservableSource<String>>() {
            @Override
            public ObservableSource<String> call() throws Exception {
                return Observable.just("defer observable instance");
            }
        });

        //定时发送事件
        intervalObservavble = Observable.interval(100, TimeUnit.MICROSECONDS);

        //发送整数序列事件，共调用count次onNext(start++)
        rangeObservable = Observable.range(1, 20);

        //延时2s发送事件(即调用onNext()方法)
        timerObservable = Observable.timer(2, TimeUnit.SECONDS);

        //重复执行n次observable事件
        repeatObservable = Observable.just("123").repeat();
    }

    /**
     * handle MissingBackPressureException(观察者发送事件的速度比观察者消费事件的速度快，导致被阻塞事件等待处理的数量超出了限制)
     */
    private void createFlowable() {
        Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {

            }
        }, BackpressureStrategy.ERROR); // Drop Latest Buffer

        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                /** No events will be sent by a {@link Publisher} until demand is signaled via this method.
                 *  重复调用，消费事件的数量
                 * */
                s.request(50);

                /**
                 * Request the {@link Publisher} to stop sending data and clean up resources.
                 * <p>
                 * Data may still be sent to meet previously signalled demand after calling cancel.
                 */
                s.cancel();
            }

            @Override
            public void onNext(String o) {

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        };

        flowable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(subscriber);

    }

    private void createObserver() {
        observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };
    }

    private void listActionType() {
        //A functional interface (callback) similar to Runnable but allows throwing a checked exception.
        Action action = new Action() {
            @Override
            public void run() throws Exception {

            }
        };

        //A functional interface (callback) that accepts a single value.
        consumer = new Consumer<String>() {
            @Override
            public void accept(String data) throws Exception {
                Log.i("COCO", data);
            }
        };

        consumer2 = new Consumer<Integer>() {
            @Override
            public void accept(Integer length) throws Exception {
                Log.i("COCO", length + "");
            }
        };

        //A functional interface (callback) that takes a value and returns another value
        function = new Function<String, Integer>() {
            @Override
            public Integer apply(String origin) throws Exception {

                return origin.length();
            }
        };

        //A functional interface that has a single cancel method
        Cancellable cancellable = new Cancellable() {
            @Override
            public void cancel() throws Exception {

            }
        };

        //A functional interface (callback) that returns true or false for the given input value.
        Predicate predicate = new Predicate<Integer>() {
            @Override
            public boolean test(Integer o) throws Exception {
                return o > 0;
            }
        };

        //A functional interface (callback) that returns a boolean value.
        BooleanSupplier booleanSupplier = new BooleanSupplier() {
            @Override
            public boolean getAsBoolean() throws Exception {
                return false;
            }
        };

        //Represents a disposable resource. 清理/处置资源
        Disposable disposable = new Disposable() {
            @Override
            public void dispose() {

            }

            @Override
            public boolean isDisposed() {
                return false;
            }
        };
    }

    private void listOperators() {
        Disposable biDisposable = justStrObservable.flatMap(new Function<List<String>, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(List<String> list) throws Exception {
                //Toast.makeText(MainActivity.this, "map operator's result___" + list.size(), Toast.LENGTH_SHORT).show();
                return Observable.fromIterable(list);
            }
        }).map(new Function<String, Integer>() {
            @Override
            public Integer apply(String s) throws Exception {
                //Log.i("COCO", "biDisposable: " + s);
                return s.length();
            }
        }).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer >= 5;
            }
        }).take(1).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i("COCO", "before calling onNext() method");
            }
        }).subscribe(consumer2);

        compositeDisposable.add(biDisposable);
    }

    private void requireScheduler() {
        observable.subscribeOn(Schedulers.computation())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void subscribeEvent() {
        observable.subscribe(observer);
        iterableObervable.subscribe(observer);

        //justObservable.subscribe(observer);
        Disposable disposable = justObservable.subscribe(consumer);
        compositeDisposable.add(disposable);
    }

    private void requestData() {
        RetrofitProxy.sendRequest();
    }


    @Override
    protected void onDestroy() {
        super.onDestroy();
        compositeDisposable.dispose();
    }
}
